package com.doit.day02;

import com.alibaba.fastjson.JSON;
import lombok.*;
import org.apache.commons.lang3.RandomStringUtils;
import org.apache.commons.lang3.RandomUtils;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.ArrayList;
import java.util.Properties;

/**
 * 生产的数据格式：  造数据
 * {"guid":1,"eventId":"pageview","timestamp":1637868346789}  isNew = 1
 * {"guid":1,"eventId":"addcard","timestamp":1637868347625}   isNew = 0
 * {"guid":2,"eventId":"collect","timestamp":16378683463219}
 * {"guid":3,"eventId":"paid","timestamp":16378683467829}
 * ......
 */
public class Exercise_Producer {
    public static void main(String[] args) throws InterruptedException {
        //封装了一些event事件信息
        ArrayList<String> list = new ArrayList<>();
        list.add("$AppStartPassively");
        list.add("$indicators_view");
        list.add("$add_bag");
        list.add("$AppInstall");
        list.add("$Claim_input");
        list.add("$Claim_detail");
        list.add("$Claim_categary");
        list.add("$AppEnd");
        list.add("$questionaire_submit_answer");
        list.add("$survey_choose");
        list.add("$iuws_searce_result");
        list.add("$app_search_result");
        list.add("$next_step_yixiang");

        Properties props = new Properties();
        props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"linux01:9092");
        props.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        props.setProperty(ProducerConfig.ACKS_CONFIG,"-1");
        //生产者对象
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(props);

        //创建一个对象
        EventLog eventLog = new EventLog();
        while (true){
            //给对象的属性设置参数值
            eventLog.setGuid(RandomUtils.nextInt(0,10000));
            eventLog.setEventId(list.get(RandomUtils.nextInt(0,list.size())));
            eventLog.setTimeStamp(System.currentTimeMillis());
            //将对象转换成json串
            String json = JSON.toJSONString(eventLog);
            //封装发送的数据
            ProducerRecord<String, String> record = new ProducerRecord<>("eventlog", json);
            //发送数据
            producer.send(record);
            producer.flush();
            //避免发送的太快，要睡一会
            Thread.sleep(RandomUtils.nextInt(0,300));
        }
    }
}


@NoArgsConstructor
@AllArgsConstructor
@Getter
@Setter
@ToString
class EventLog{
    private Integer guid;
    private String eventId;
    private long timeStamp;
    private int isNew = 1;
}
